package mapreduce;

import java.io.File;
import java.io.IOException;
import java.util.Vector;

import org.apache.commons.collections.map.MultiValueMap;

public class ReduceTaskConf<KEYIN, VALUEIN, KEYOUT, VALUEOUT> extends TaskConf {

	private String FileDir;
	private Class<?> ReduceClass;
	private Class<?> ReduceInputKeyClass;
	private Class<?> ReduceInputValueClass;
	private Manager<KEYOUT, VALUEOUT> manager;
	public ReduceTaskConf(Class<?> ReduceClass, Class<?> inputkeyclass,
			Class<?> inputvalueclass, String Jobdir, long buffersize, int Taskid, boolean mflag, Vector<pair<KEYIN, VALUEIN>>[][] map ,Vector< pair< KEYOUT, VALUEOUT>>[] resultmap)
			throws IOException {
		super(Taskid, mflag);
		File reduceresultdir = new File(Jobdir + "reduce/output/");
		this.FileDir = Jobdir + "reduce/input/" + Taskid + "/";
		this.ReduceClass = ReduceClass;
		if (!reduceresultdir.exists())
			reduceresultdir.mkdirs();
		if(mflag==false)
			manager = new ReduceDiskManager<KEYOUT, VALUEOUT>(Jobdir, buffersize,
				Taskid);
		else {
			manager = new ReduceBufferManager<KEYIN, VALUEIN, KEYOUT, VALUEOUT>(Taskid , map, resultmap);
			
		}
		this.ReduceInputKeyClass = inputkeyclass;
		this.ReduceInputValueClass = inputvalueclass;
	}

	public String getFileDir() {
		return this.FileDir;
	}

	public Class<?> getReduceClass() {
		return this.ReduceClass;
	}

	/**
	 * <p>
	 * write out the key and value.
	 * </p>
	 * 
	 * @param key
	 * @param value
	 * @throws IOException
	 */
	public void write(KEYOUT key, VALUEOUT value) throws IOException {
		manager.write(key, value);

	}

	public void close() throws IOException {
		manager.close();
	}

	public Class<?> getInputKeyClass() {
		return this.ReduceInputKeyClass;
	}

	public Class<?> getInputValueClass() {
		return this.ReduceInputValueClass;
	}

}
